//
//  SocketChannel.m
//  socket
//
//  Created by wjxfhxy on 14-01-30.
//  Copyright 2011 GuangZhouXuanWu. All rights reserved.
//

#import "SocketChannel.h"

#import "RequestChannelPool.h"

#import "MemoryStream.h"

#import "BitConvert.h"

#import <sys/socket.h>
#import <netinet/in.h>
#import <arpa/inet.h>

#import "ErrorCode.h"

#define SC_TIME_OUT 30

@interface CSocketChannel ()
{
	NSString* _IP;
	NSUInteger _port;
    
    CMemoryStream* _request;
    CMemoryStream* _response;
    
    NSInteger _writeStep;
	NSInteger _readStep;
	
	CFReadStreamRef _read;
	CFWriteStreamRef _write;
 	
	NSTimer* _sendTimer;
	NSTimer* _recTimer;
	NSTimer* _connectTimer;
	
	__unsafe_unretained id<CSocketChannelDelegate> _delegate;
    
    NSLock* m_lock;
    
    NSTimeInterval _sendTimeout;
    NSTimeInterval _recvTimeout;
    
    XMSDispatchQueueWrapper *_delegatequeue;
}

@end

extern void ReadStreamCallBack(CFReadStreamRef stream,CFStreamEventType type,void *info);
extern void WriteStreamCallBack(CFWriteStreamRef stream,CFStreamEventType type,void *info);

@implementation CSocketChannel

@synthesize IP = _IP;
@synthesize port = _port;

@synthesize request = _request;
@synthesize response = _response;

@synthesize delegate = _delegate;

@synthesize sendTimeout = _sendTimeout;
@synthesize recvTimeout = _recvTimeout;

-(id)initWithIP:(NSString*)szIP port:(NSInteger)nPort
{
	self = [super init];
    
	_IP=[szIP retain_mm];
	_port=nPort;
	
	_read=NULL;
	_write=NULL;
    
    _sendTimeout=SC_TIME_OUT;
    
    _recvTimeout=SC_TIME_OUT;
    
    _delegatequeue=[[XMSDispatchQueueWrapper alloc]initWithName:""];
    
    self.dispatchQueueWrapper=[[[XMSDispatchQueueWrapper alloc]initWithName:""]autorelease_mm];
    
	return self;
}

-(void)dealloc
{
	[_IP release_mm];
	
	_delegate=nil;
    
    [_request release_mm];
    [_response release_mm];
    
    [_delegatequeue release_mm];
    
	superdealloc_mm;
}

-(void)setDelegate:(id<CSocketChannelDelegate>)delegate
{
    [_delegatequeue dispatchThreadSync:^
    {
        _delegate = delegate;
    }];
}

-(BOOL)createSocket
{
    CFStreamCreatePairWithSocketToHost(kCFAllocatorDefault,(CFStringRef)_IP, ToUInt _port,&_read,&_write);
    if(NULL!=_read&&NULL!=_write)
        return YES;
    return NO;
}

-(BOOL)initSocketChannel
{
//    [self dispatchQueueRunType:EDispatchQueueRunTypeAsync checkTerminate:NO asyncComplete:^
//    {
        BOOL bResult=NO;
        [self createSocket];
        if(NULL!=_read&&NULL!=_write)
        {
            CFReadStreamSetProperty(_read,kCFStreamPropertyShouldCloseNativeSocket,kCFBooleanTrue);
            CFWriteStreamSetProperty(_write,kCFStreamPropertyShouldCloseNativeSocket,kCFBooleanTrue);
            
            CFSocketContext theContext={0};
            theContext.info=self;
            CFOptionFlags readevents=kCFStreamEventHasBytesAvailable|kCFStreamEventErrorOccurred|kCFStreamEventEndEncountered|kCFStreamEventOpenCompleted;
            if(!CFReadStreamSetClient(_read,readevents,(CFReadStreamClientCallBack)&ReadStreamCallBack,(CFStreamClientContext *)(&theContext)))
            {
                goto FUNEXIT;
            }
            
            CFOptionFlags writeevents=kCFStreamEventCanAcceptBytes|kCFStreamEventErrorOccurred|kCFStreamEventEndEncountered|kCFStreamEventOpenCompleted;
            if(!CFWriteStreamSetClient(_write,writeevents,(CFWriteStreamClientCallBack)&WriteStreamCallBack,(CFStreamClientContext *)(&theContext)))
            {
                goto FUNEXIT;
            }
                    
//            if(YES!=[self.channelPool addRequestChannelPoolRunLoop:^(NSRunLoop *runloop)
//            {
//                _connectTimer=[[NSTimer timerWithTimeInterval:SC_TIME_OUT target:self selector:@selector(connectTimeOut) userInfo:nil repeats:NO]retain];
//                CFRunLoopAddTimer([runloop getCFRunLoop],(CFRunLoopTimerRef)_connectTimer,kCFRunLoopDefaultMode);
//                
//                CFReadStreamScheduleWithRunLoop(_read,[runloop getCFRunLoop],kCFRunLoopDefaultMode);
//                CFWriteStreamScheduleWithRunLoop(_write,[runloop getCFRunLoop],kCFRunLoopDefaultMode);
//            }])
//                goto FUNEXIT;
            
            if(YES!=CFWriteStreamOpen(_write))
                goto FUNEXIT;
            if(YES!=CFReadStreamOpen(_read))
                goto FUNEXIT;
            
            bResult=YES;
        }
    FUNEXIT:
//        if(bResult==NO)
//        {
//            [_delegatequeue dispatchThreadAsync:^
//            {
//                [_delegate socketOperateResult:self type:ESocketOperateResultTypeError result:bResult];
//            }];
//        }
//    }];
    return bResult;
}

-(void)_closeSocket
{
    [self invalidateConnectTimer];
    [self invalidateSendTimer];
    [self invalidateRecvTimer];
    
    if(_write!=NULL)
    {
        [self closeWriteStreamCallBack];
        CFWriteStreamClose(_write);
        CFReleaseRef(_write);
    }
    if(_read!=NULL)
    {
        [self closeReadStreamCallBack];
        CFReadStreamClose(_read);
        CFReleaseRef(_read);
    }
}

-(void)closeSocket
{
    [self dispatchQueueRunType:EDispatchQueueRunTypeSync checkTerminate:NO asyncBlock:^
    {
        [self _closeSocket];
    }];
}

-(void)cancelRequest
{
    [super terminateRequest];
}

-(BOOL)terminateSelfRequest:(BOOL)bRejectrequest
{
    [self dispatchQueueRunType:EDispatchQueueRunTypeAsync checkTerminate:NO asyncBlock:^
    {
        [self _closeSocket];
    }];
    [_delegatequeue dispatchThreadSync:^
    {
        [_delegate socketOperateResult:self type:ESocketOperateResultTypeError result:TERMINATEREADWRITE];
    }];
    return YES;
}

-(void)closeWriteStreamCallBack
{
	if(_write!=NULL)
	{
        [self.channelPool removeRequestChannelPoolRunLoop:^(NSRunLoop *runloop)
        {
            CFWriteStreamUnscheduleFromRunLoop(_write,[runloop getCFRunLoop],kCFRunLoopDefaultMode);
        }];
	}
}

-(void)closeReadStreamCallBack
{
	if(_read!=NULL)
	{
        [self.channelPool removeRequestChannelPoolRunLoop:^(NSRunLoop *runloop)
        {
            CFReadStreamUnscheduleFromRunLoop(_read,[runloop getCFRunLoop],kCFRunLoopDefaultMode);
        }];
	}
}

-(void)connectTimeOut
{
    [self dispatchQueueRunType:EDispatchQueueRunTypeAsync checkTerminate:NO asyncBlock:^
    {
        [self _closeSocket];
    
        [_delegatequeue dispatchThreadAsync:^
        {
            [_delegate socketOperateResult:self type:ESocketOperateResultTypeConnectTimeout result:SOCKETCONNECTTIMEOUT];
        }];
    }];
}

-(BOOL)initConnectTimer
{
    __block BOOL bResult=NO;
    [self.channelPool addRequestChannelPoolRunLoop:^(NSRunLoop *runloop)
    {
        if(nil==_connectTimer)
        {
            _connectTimer=[[NSTimer timerWithTimeInterval:SC_TIME_OUT target:self selector:@selector(connectTimeOut) userInfo:nil repeats:NO]retain_mm];
            [runloop addTimer:_connectTimer forMode:NSDefaultRunLoopMode];
            bResult=YES;
        }
    }];
    return bResult;
}

-(void)invalidateConnectTimer
{
	if(nil!=_connectTimer)
	{
		[_connectTimer invalidate];
        ReleaseObj(_connectTimer);
	}
}

-(void)initSendTimer
{
    [self.channelPool addRequestChannelPoolRunLoop:^(NSRunLoop *runloop) 
    {
        if(nil==_sendTimer)
        {
            _sendTimer=[[NSTimer timerWithTimeInterval:_sendTimeout target:self selector:@selector(sendTimerTimeOut) userInfo:nil repeats:NO]retain_mm];
            [runloop addTimer:_sendTimer forMode:NSDefaultRunLoopMode];
        }
    }];
}

-(void)invalidateSendTimer
{
	if(nil!=_sendTimer)
	{
		[_sendTimer invalidate];
        ReleaseObj(_sendTimer);
	}
}

-(void)initRecvTimer
{
    //[m_socketchannelpool.m_curretnrunloop addTimer:_recTimer forMode:NSDefaultRunLoopMode];
    [self.channelPool addRequestChannelPoolRunLoop:^(NSRunLoop *runloop)
    {
        if(nil==_recTimer)
        {
            _recTimer=[[NSTimer timerWithTimeInterval:_recvTimeout target:self selector:@selector(recvTimerTimeOut) userInfo:nil repeats:NO]retain_mm];
            [runloop addTimer:_recTimer forMode:NSDefaultRunLoopMode];
        }
    }];
}

-(void)invalidateRecvTimer
{
	if(nil!=_recTimer)
	{
		[_recTimer invalidate];
        ReleaseObj(_recTimer);
	}
}


-(void)sendTimerTimeOut
{
    [self dispatchQueueRunType:EDispatchQueueRunTypeAsync checkTerminate:NO asyncBlock:^
    {
        [self _closeSocket];
    
        [_delegatequeue dispatchThreadAsync:^
        {
            [_delegate socketOperateResult:self type:ESocketOperateResultTypeWriteTimeout result:WRITESTREAMTIMEOUT];
        }];
    }];
}

-(void)recvTimerTimeOut
{
    [self dispatchQueueRunType:EDispatchQueueRunTypeAsync checkTerminate:NO asyncBlock:^
    {
        [self _closeSocket];

        [_delegatequeue dispatchThreadAsync:^
        {
            [_delegate socketOperateResult:self type:ESocketOperateResultTypeReadTimeout result:READSTREAMTIMEOUT];
        }];
    }];
}

-(NSData*)getSendData
{
	NSData* data=nil;
	switch(_writeStep)
	{
		case 0:
		{
			[_request setPosition:0];
			NSInteger n = [_request length];
            NSUInteger nLen = 0;
            CMemoryStream* m=[[CMemoryStream new]autorelease_mm];
            [CBitConvert toIntArray: ToUInt n + 4 inoutput:m output:&nLen];
            data=[[m.data retain_mm]autorelease_mm];
			break;
		}
		case 1:
		{
			NSMutableData* tmp=_request.data;
			data=[tmp subdataWithRange:NSMakeRange([_request position],MIN([_request length]-[_request position],BUFSIZ*100))];
			break;
		}
	}
	return data;
}

-(NSInteger)getRecvDatalength
{
	NSInteger nLength=0;
	switch(_readStep)
	{
		case 0:
		{
			nLength=4;
			break;
		}
		case 1:
		{
			nLength=[_response position]-[_response length];
			break;
		}
	}
	return nLength;
}

-(void)writeStream:(CFWriteStreamRef)stream evntType:(CFStreamEventType)type
{
    [self dispatchQueueRunType:EDispatchQueueRunTypeAsync checkTerminate:YES asyncBlock:^
    {
        if(_write==NULL)
            return ;
        if(_writeStep==2)
            return;
        switch((NSInteger)type)
        {
            case kCFStreamEventOpenCompleted:
            {
                [self invalidateConnectTimer];
                [self initSendTimer];
                break;
            }
            case kCFStreamEventCanAcceptBytes:
            {
                [self invalidateSendTimer];
                
                NSData* senddata=[self getSendData];
                if(senddata!=nil)
                {
                    NSInteger nSendedlen=0;
                    NSInteger nResutl=[self writeStream:(char*)senddata.bytes length:senddata.length output:&nSendedlen];
                    switch(nResutl)
                    {
                        case TRUE:
                        {
                            ECloseSocketOperateType type=ECloseSocketOperateTypeNone;
                            
                            switch(_writeStep)
                            {
                                case 0:
                                {
                                    [_delegatequeue dispatchThreadAsync:^
                                    {
                                        [_delegate socketChannel:self willSendTotalBytes:senddata.length];
                                    }];
                                    _writeStep=1;
                                    break;
                                }
                                case 1:
                                {
                                    [_delegatequeue dispatchThreadAsync:^
                                    {
                                        [_delegate socketChannel:self hadSendBytes:nSendedlen];
                                    }];
                                    [_request setPosition:[_request position]+nSendedlen];
                                    if([_request isEndPosition])
                                    {
                                        _writeStep=2;
                                        type=ECloseSocketOperateTypeSendFinish;
                                    }
                                    break;
                                }
                            }
                            
                            switch(type)
                            {
                                case ECloseSocketOperateTypeNone:
                                {
                                    if(YES!=CFWriteStreamCanAcceptBytes(_write))
                                    {
                                        [self initSendTimer];
                                    }
                                    break;
                                }
                                case ECloseSocketOperateTypeSendFinish:
                                {
                                    [self closeWriteStreamCallBack];
                                    if(2==_writeStep)
                                        [self initRecvTimer];
                                    break;
                                }
                                default:
                                {
                                    [self _closeSocket];
                                    break;
                                }
                            }
                            
                            break;
                        }
                        case WRITESTREAMERROR:
                        {
                            [self _closeSocket];
                            
                            [_delegatequeue dispatchThreadAsync:^
                            {
                                [_delegate socketOperateResult:self type:ESocketOperateResultTypeError result:WRITESTREAMERROR];
                            }];
                            
                            break;
                        }
                        case WRITESTREAMEND:
                        {
                            break;
                        }
                    }
                }
                
                break;
            }
            case kCFStreamEventErrorOccurred:
            {
                [self _closeSocket];
                
                [_delegatequeue dispatchThreadAsync:^
                {
                    [_delegate socketOperateResult:self type:ESocketOperateResultTypeError result:WRITESTREAMERROR];
                }];
                break;
            }
        }
    }];
}

-(void)readStream:(CFReadStreamRef)stream evntType:(CFStreamEventType)type
{
	[self dispatchQueueRunType:EDispatchQueueRunTypeAsync checkTerminate:YES asyncBlock:^
    {
        if(_read==NULL)
            return ;
        if(_readStep==2)
            return;
        switch((NSInteger)type)
        {
            case kCFStreamEventOpenCompleted:
            {
                [self invalidateConnectTimer];
                break;
            }
            case kCFStreamEventHasBytesAvailable:
            {
                [self invalidateRecvTimer];
                
                NSInteger nRecvedlen=0;
                NSInteger nLen=[self getRecvDatalength];
                if(nLen>0)
                {
                    char* buf=(char*)malloc(sizeof(char)*nLen);
                    NSInteger nResult=[self readStream:buf length:nLen output:&nRecvedlen];
                    NSData* data = [[[NSData alloc]initWithBytesNoCopy:buf length:nRecvedlen freeWhenDone:YES]autorelease_mm];
                    switch(nResult)
                    {
                        case TRUE:
                        {
                            ECloseSocketOperateType type=ECloseSocketOperateTypeNone;
                            
                            switch(_readStep)
                            {
                                case 0:
                                {
                                    _readStep=1;
                                    [_response setPosition:[CBitConvert fromIntBuffer:(char*)data.bytes]-4];
                                    break;
                                }
                                case 1:
                                {
                                    [_response appendData:data];
                                    if([_response isEndPosition])
                                    {
                                        [_response setPosition:0];
                                        _readStep=2;
                                        type=ECloseSocketOperateTypeRecvFinish;
                                    }
                                    break;
                                }
                            }
                            
                            switch(type)
                            {
                                case ECloseSocketOperateTypeNone:
                                {
                                    if(YES!=CFReadStreamHasBytesAvailable(_read))
                                    {
                                        [self initRecvTimer];
                                    }
                                    break;
                                }
                                case ECloseSocketOperateTypeRecvFinish:
                                {
                                    [self closeReadStreamCallBack];
                                    
                                    [_delegatequeue dispatchThreadAsync:^
                                    {
                                        [_delegate socketOperateResult:self type:ESocketOperateResultTypeFinish result:TRUE];
                                    }];
                                    break;
                                }
                                default:
                                {
                                    [self _closeSocket];
                                    break;
                                }
                            }
                            break;
                        }
                        case READSTREAMERROR:
                        {
                            [self _closeSocket];
                            [_delegatequeue dispatchThreadAsync:^
                            {
                                [_delegate socketOperateResult:self type:ESocketOperateResultTypeError result:READSTREAMERROR];
                            }];
                            break;
                        }
                        case READSTREAMEND:
                        {
                            break;
                        }
                    }
                }
                break;
            }
            case kCFStreamEventErrorOccurred:
            {
                [self _closeSocket];
                [_delegatequeue dispatchThreadAsync:^
                {
                    [_delegate socketOperateResult:self type:ESocketOperateResultTypeError result:READSTREAMERROR];
                }];
                break;
            }
        }	
	}];
}

-(NSInteger)writeStream:(char*)buf length:(NSInteger)nLength output:(NSInteger*)nSendedlen
{
	NSInteger nResult=TERMINATEREADWRITE;
	
    @autoreleasepool
    {
        nResult=CFWriteStreamCanAcceptBytes(_write);
        if(YES==nResult)
        {
            NSInteger nSend=CFWriteStreamWrite(_write,(UInt8*)buf,nLength);
            if(nSend>0)
            {
                *nSendedlen=nSend;
                nResult=TRUE;
            }
            else if(nSend<0)
            {
                nResult=WRITESTREAMERROR;
            }
            else if(0==nSend)
            {
                if(CFWriteStreamGetStatus(_write)==kCFStreamStatusAtEnd)
                {
                    nResult=WRITESTREAMEND;
                }
                nResult=WRITESTREAMERROR;
            }
        }
        else
        {
            nResult=TERMINATEREADWRITE;
        }
    }
	
	return nResult;
}

-(NSInteger)readStream:(char*)buf length:(NSInteger)nLength output:(NSInteger*)nRecvedlen
{
	NSInteger nResult=TERMINATEREADWRITE;
    
    @autoreleasepool
    {
        nResult=CFReadStreamHasBytesAvailable(_read);
        if(YES==nResult)
        {
            NSInteger nLen=CFReadStreamRead(_read,(UInt8*)buf,nLength);
            if(nLen>0)
            {
                *nRecvedlen=nLen;
                nResult=TRUE;
            }
            else if(nLen<0)
            {
                nResult=READSTREAMERROR;
            }
            else if(0==nLen)
            {
                if(CFReadStreamGetStatus(_read)==kCFStreamStatusAtEnd)
                {
                    nResult=READSTREAMEND;
                }
                nResult=READSTREAMERROR;
            }
        }
        else
        {
            nResult=TERMINATEREADWRITE;
        }
    }
	
	return nResult;
}

-(NSInteger)initLongConnectSocketChannel
{
    [self createSocket];
    if(NULL!=_read&&NULL!=_write)
    {
        CFReadStreamSetProperty(_read,kCFStreamPropertyShouldCloseNativeSocket,kCFBooleanTrue);
        
        CFWriteStreamSetProperty(_write,kCFStreamPropertyShouldCloseNativeSocket,kCFBooleanTrue);
        
        CFSocketContext theContext={0};
        theContext.info=self;
        CFOptionFlags readevents=kCFStreamEventHasBytesAvailable|kCFStreamEventErrorOccurred|kCFStreamEventEndEncountered|kCFStreamEventOpenCompleted;
        if(!CFReadStreamSetClient(_read,readevents,(CFReadStreamClientCallBack)&ReadStreamCallBack,(CFStreamClientContext *)(&theContext)))
        {
            return NO;
        }
        
        CFOptionFlags writeevents=kCFStreamEventCanAcceptBytes|kCFStreamEventErrorOccurred|kCFStreamEventEndEncountered|kCFStreamEventOpenCompleted;
        if(!CFWriteStreamSetClient(_write,writeevents,(CFWriteStreamClientCallBack)&WriteStreamCallBack,(CFStreamClientContext *)(&theContext)))
        {
            return NO;
        }
        if(YES!=CFWriteStreamOpen(_write))
        {
            return NO;
        }
        if(YES!=CFReadStreamOpen(_read))
        {
            return NO;
        }
    }
    return YES;
}

-(BOOL)beginRequest
{
    _writeStep=0;
    _readStep=0;
    if(YES!=[self.channelPool addRequestChannelPoolRunLoop:^(NSRunLoop *runloop)
             {
                 _connectTimer=[[NSTimer timerWithTimeInterval:SC_TIME_OUT target:self selector:@selector(connectTimeOut) userInfo:nil repeats:NO]retain];
                 CFRunLoopAddTimer([runloop getCFRunLoop],(CFRunLoopTimerRef)_connectTimer,kCFRunLoopDefaultMode);
                 
                 CFReadStreamScheduleWithRunLoop(_read,[runloop getCFRunLoop],kCFRunLoopDefaultMode);
                 CFWriteStreamScheduleWithRunLoop(_write,[runloop getCFRunLoop],kCFRunLoopDefaultMode);
             }])
    {
        return NO;
    }
//    if(YES!=CFWriteStreamOpen(_write))
//    {
//        CFStreamStatus n = CFWriteStreamGetStatus(_write);
//        CFStreamError e = CFWriteStreamGetError(_write);
//        return NO;
//    }
//    if(YES!=CFReadStreamOpen(_read))
//    {
//        return NO;
//    }
    return YES;
}

-(BOOL)endRequest
{
    [self closeWriteStreamCallBack];
    [self closeReadStreamCallBack];
//    if(_write!=NULL)
//        CFWriteStreamClose(_write);
//    if(_read!=NULL)
//        CFReadStreamClose(_read);
    return YES;
}

@end

void ReadStreamCallBack(CFReadStreamRef stream, CFStreamEventType type, void *info)
{
    @autoreleasepool
    {
        CSocketChannel* sc=[[(CSocketChannel*)info retain_mm]autorelease_mm];
        [sc readStream:stream evntType:type];
	}
}

void WriteStreamCallBack(CFWriteStreamRef stream, CFStreamEventType type, void *info)
{
	@autoreleasepool
    {
        CSocketChannel* sc=[[(CSocketChannel*)info retain_mm]autorelease_mm];
        [sc writeStream:stream evntType:type];
    }
}

